package com.example.hometown_walker.service.process;

import com.example.hometown_walker.util.CommonUtil;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import lombok.SneakyThrows;
import org.apache.commons.lang3.time.DateFormatUtils;

import java.text.DecimalFormat;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.UUID;

/**
 * @description: 监控线程、Manager（可拆分为两个类）
 * @author: chenchaoran
 * @date: 2020-04-01
 */
public class ProcessManager extends Thread {
//    private static final ThreadPoolExecutor exe = new ThreadPoolExecutor(5, 5, 60, TimeUnit.SECONDS, new LinkedBlockingDeque<>());
    private static final DecimalFormat df = new DecimalFormat("0.00");
    private static final long TimeLimit = 1000 * 30;
    /**
     * 缓存各任务组进度
     */
    private static final Map<String, Float> CacheMap = Maps.newHashMap();

    private String taskId;// 标记本次任务
    private Map<String, TaskThread> taskThreads;
    private int expectedProcess;// 预期总进度 = 100 * 任务数量
    private int totalProcess = 0;

    private ProcessManager(String taskId, List<ProcessTask> tasks) {
        this.taskId = taskId;
        this.expectedProcess = 100 * tasks.size();
        this.taskThreads = Maps.newHashMapWithExpectedSize(tasks.size());
        for (ProcessTask task : tasks) {
            TaskThread thread = new TaskThread(task);
            this.taskThreads.put(thread.getName(), thread);
        }
    }

    private void manage() {
        if (taskThreads != null) {
            taskThreads.forEach((x, y) -> y.start());
        }
    }

    @SneakyThrows
    @Override
    public void run() {
        System.out.println("--------进度统计线程开启...（期望总进度值："+ this.expectedProcess + "）");
        long start = System.currentTimeMillis();
        // 统计各线程进度
        while (this.totalProcess < this.expectedProcess) {
            Thread.sleep(1000);//1秒后开始统计，并每隔1秒统计一次
            gatherAllProcess();
            CacheMap.put(this.taskId, (float) this.totalProcess / this.expectedProcess);

            if (isTaskTimeOut(start)) {
                return;
            }
        }
        System.out.println("--------全部任务执行完成，进度统计结束！");
    }

    // 汇总任务进度
    private void gatherAllProcess() {
        int sum = 0;
        for (TaskThread thread : this.taskThreads.values()) {
            sum += thread.getProcess();
            if (thread.isAlive()) {
                System.out.println(thread.getTaskName() + "-进度：" + thread.getProcess() + "% .");
            } else {
                System.out.println(thread.getName() + " not alive");
            }
        }
        this.totalProcess = sum;
        System.out.println("--当前总进度值：" + this.totalProcess + "\n");
    }

    // 进度监控超时处理（改方法待优化）
    // 通过interrupt()方法的调用，只能将线程标记为待中断，在一个合适的时机(?)才会结束线程，或者对处于wait,sleep等阻塞状态的线程抛出InterruptedException异常。
    // interrupt()无法中断非InterruptedException阻塞。
    private boolean isTaskTimeOut(long start) {
        long current = System.currentTimeMillis();
        if (current - start > TimeLimit) {
            // 超过10分钟仍未执行完毕，清除map,中断线程
            System.out.println("超时（10分钟）警告！终结任务。");
            CacheMap.remove(this.taskId);
            this.taskThreads.values().forEach(x -> {
                if (x.isAlive()) {
                    x.interrupt();// 处于sleeping的线程，执行interrupt()方法，会抛出'InterruptedException'
                    System.out.println(x.getName() + "强制中断~~");
                }
            });
            return true;
        }
        return false;
    }

    /**
     * 提交任务
     *
     * @param tasks 任务组
     * @return taskId 本次任务id，用于查询进度
     */
    public static String submit(List<ProcessTask> tasks) {
        String taskId = UUID.randomUUID().toString() + DateFormatUtils.format(new Date(), "MMddHHmmss");
        System.out.println(taskId);
        CacheMap.put(taskId, 0f);
        ProcessManager manager = new ProcessManager(taskId, tasks);
        manager.start();// 监控线程启动
        manager.manage();// 任务组执行
        return taskId;
    }

    /**
     * 查询任务进度
     *
     * @param taskId 任务组id
     * @return int 进度
     * @throws Exception id不存在
     */
    public static float getProcess(String taskId) throws Exception {
        Float process = CacheMap.get(taskId);
        if (process != null) {
            if (process == 1) {
                CacheMap.remove(taskId);
            }
            return process;
        }
        throw new Exception("当前任务id不存在");
    }


    public static class TaskThread extends Thread {
        private ProcessTask task;

        public TaskThread(ProcessTask task) {
            super(task);
            this.task = task;
        }

        public int getProcess() {
            return task.getProcess();
        }

        public String getTaskName() {
            return task.taskName();
        }
    }


    public static void main(String[] args) {
        // 模拟两个耗时任务
        ProcessTask task1 = new ProcessTask() {

            @Override
            String taskName() {
                return "生成PDF文件";
            }

            @SneakyThrows
            @Override
            public void doTask() {
                Thread.sleep(2000);
                setProcess(30);
                Thread.sleep(2000);
                setProcess(75);
                Thread.sleep(5000);
                setProcess(100);
            }
        };

        ProcessTask task2 = new ProcessTask() {

            @Override
            String taskName() {
                return "生成WORD文件";
            }

            @SneakyThrows
            @Override
            public void doTask() {
                Thread.sleep(5000);
                setProcess(50);

                Thread.sleep(5000);
                setProcess(80);

//                Thread.sleep(30 * 1000); // interrupt()可以中断sleep()阻塞状态，并抛出异常

                // 执行一个耗时任务，30s
                CommonUtil.timingTest(); // interrupt()无法中断改过程，改耗时过程会一直执行完，线程才结束。
                setProcess(100);
            }
        };

        // 提交任务，并且在任务未全部完成前，不断查询进度
        String taskId = ProcessManager.submit(Lists.newArrayList(task1, task2));
        while (true) {
            float totalProcess = 0;
            try {
                totalProcess = ProcessManager.getProcess(taskId);
            } catch (Exception e) {
                System.out.println(e.getMessage());
                break;
            }
            System.out.println(String.format("=======> 当前总进度：%s%%.\n\n", df.format(totalProcess * 100)));

            if (totalProcess >= 1) {
                break;
            }

            try {
                Thread.sleep(2000);// 每隔一秒查询一次进度
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

}
